package mr;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;

import java.util.ArrayList;
import java.util.Hashtable;
import java.util.Map.Entry;
import java.util.StringTokenizer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import util.ScanList;

public class SecondMapper extends
		Mapper<LongWritable, Text, Text, NullWritable> {

	//private ArrayList<Integer> scanList;
	//private Hashtable<Integer, Integer> keyIndex;
	private static final Log LOG = LogFactory.getLog(SecondMapper.class);

	private long totalsum =0;
	private ScanList<Long> psums;
	private long startLineKey = -1;
	
	protected void setup(Context context) throws IOException,
			InterruptedException {
		
		Path[] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration()); //context.getLocalCacheFiles();
		File f = new File(paths[0].toString());
		// Files.readAllLines(new
		// java.nio.file.Path(paths[0].toString()),StandardCharsets.UTF_8);
		//Charset charset = StandardCharsets.UTF_8;
		String partialSumList = "";
		long start = System.currentTimeMillis();
		BufferedReader reader = new BufferedReader(new FileReader(paths[0].toString())); //Files.newBufferedReader(f.toPath(),charset);
		try{

			String line = null;
			while ((line = reader.readLine()) != null) {
				partialSumList += line;
			}
			partialSumList.trim();
			//System.out.println("**************"+partialSumList);
			//take care, maybe very large
			//LOG.debug("**************"+partialSumList);
		} catch (IOException x) {
			System.err.format("IOException: %s%n", x);
		}

		LOG.debug("finish reading partialSumList. Time =" + (System.currentTimeMillis()-start));
		
		start = System.currentTimeMillis();
		if (partialSumList.length() > 0) {
			String[] tmp = partialSumList.split(";");
			psums = new ScanList<Long>();
			long sum = 0;
			for (String val : tmp) {
				if (val != null && val.length() > 0) {
					long key = Long.parseLong(val.split("=")[0]);
					long parSum = Integer.parseInt(val.split("=")[1]);
					sum += parSum;
					psums.put(key, sum); //scan
				}
			}
		}
		LOG.debug("finish parsing partialSumList. Time =" + (System.currentTimeMillis()-start));

	}
	
	public void map(LongWritable key, Text list, Context context)
			throws IOException, InterruptedException {
		if (startLineKey <0) 
			startLineKey = key.get();
		
		if (psums.size() > 0) {
			long start = System.currentTimeMillis();
			
			//int idx = key.get();
			//totalsumlong sum = 0;
			Entry<Long, Long> de = psums.floorEntry(startLineKey);
			if(de != null && de.getKey() != startLineKey)
				totalsum = de.getValue();
			
			StringBuffer newList = new StringBuffer(2048);
			
			 
			StringTokenizer itr = new StringTokenizer(list.toString().trim());
			while (itr.hasMoreTokens()) {
				totalsum += Integer.parseInt(itr.nextToken());
				newList.append(totalsum +" ");
			}
			
//			String[] eles = list.toString().trim().split(" ");
//			for(String str: eles){
//				newList.append(sum+" ");
//				sum += Integer.parseInt(str);
//			}
			
			LOG.debug("finish map scanList. Time =" + (System.currentTimeMillis()-start));
			context.write(new Text(newList.toString()), NullWritable.get());
			
		}

	}
}
